package com.mimaxueyuan.disruptor.demo6;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

/**
 * @author: Kevin
 */
public class LongEventMain6 {

	public static long startTime;

	@SuppressWarnings("unchecked")
	public static void main(String[] args) throws InterruptedException {
		// 初始化线程池-用户执行Consumer
		// Executor threadFactory = Executors.newCachedThreadPool();
		ThreadFactory threadFactory = Executors.defaultThreadFactory();
		// 初始化EventFactory
		LongEventFactory factory = new LongEventFactory();
		// 初始化RingBuffer的大小,必须是2的指数
		int bufferSize = 1024;
		// 初始化Disruptor
		// 默认配置
		BlockingWaitStrategy waitStrategy = new BlockingWaitStrategy(); //性能状态
		//SleepingWaitStrategy waitStrategy = new SleepingWaitStrategy();
		//YieldingWaitStrategy waitStrategy = new YieldingWaitStrategy();
		Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(factory, bufferSize, threadFactory, ProducerType.MULTI, waitStrategy);
		
		LongEventHandler61[] arr1 = new LongEventHandler61[10]; 
		for(int i=0;i<10;i++) {
			arr1[i] = new LongEventHandler61();
		}
		LongEventHandler62[] arr2 = new LongEventHandler62[10]; 
		for(int i=0;i<10;i++) {
			arr2[i] = new LongEventHandler62();
		}
		LongEventHandler63[] arr3 = new LongEventHandler63[10]; 
		for(int i=0;i<10;i++) {
			arr3[i] = new LongEventHandler63();
		}
		LongEventHandler64[] arr4 = new LongEventHandler64[10]; 
		for(int i=0;i<10;i++) {
			arr4[i] = new LongEventHandler64();
		}
		
		//[ringbuffer] -> LongEventHandler61[多线程]
		disruptor.handleEventsWithWorkerPool(arr1);
		
		//[ringbuffer] -> LongEventHandler61[多线程] -> LongEventHandler62[单线程]
//		disruptor.handleEventsWithWorkerPool(arr1).handleEventsWith(new LongEventHandler62());
		
		//[ringbuffer] -> LongEventHandler61[多线程] -> LongEventHandler62[多线程]
//		disruptor.handleEventsWithWorkerPool(arr1).handleEventsWithWorkerPool(arr2);
		
		//		-> LongEventHandler61[多线程]
		//[ringbuffer]  
		//		-> LongEventHandler62[多线程]
//		disruptor.handleEventsWithWorkerPool(arr1);
//		disruptor.handleEventsWithWorkerPool(arr2);
		
		//		-> LongEventHandler61[多线程]
		//[ringbuffer] 		-> LongEventHandler63[多线程]
		//		-> LongEventHandler62[多线程]
//		EventHandlerGroup<LongEvent> g1 = disruptor.handleEventsWithWorkerPool(arr1);
//		EventHandlerGroup<LongEvent> g2 = disruptor.handleEventsWithWorkerPool(arr2);
//		g1.and(g2).thenHandleEventsWithWorkerPool(arr3);
		
		//		-> LongEventHandler61[多线程]	-> LongEventHandler62[多线程]
		//[ringbuffer] 													-> LongEventHandler61[多线程]
		//		-> LongEventHandler63[多线程]	-> LongEventHandler64[多线程]
//		EventHandlerGroup<LongEvent> g1 = disruptor.handleEventsWithWorkerPool(arr1).handleEventsWithWorkerPool(arr2);
//		EventHandlerGroup<LongEvent> g2 = disruptor.handleEventsWithWorkerPool(arr3).thenHandleEventsWithWorkerPool(arr4);
//		g1.and(g2).thenHandleEventsWithWorkerPool(arr1); //多线程,多个handler处理的数据不重复
//		g1.and(g2).handleEventsWith(arr1); //多线程,多个handler处理的数据重复
		
		// 开启Disruptor,开启所有线程
		disruptor.start();
		// 获取RingBuffer
		RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
		// 采用固定模式向RingBuffer中添加事件
		final LongEventProducer producer = new LongEventProducer(ringBuffer);

		// 10线程并发下的Disruptor
		ExecutorService executor1 = Executors.newCachedThreadPool();
		startTime = System.currentTimeMillis();
		for (long i = 0; i < 2; i++) {
			final long vv = i;
			executor1.submit(new Runnable() {
				public void run() {
					try {
						producer.onData(vv);
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			});
		}

		// 停止所有Consumer
		// disruptor.halt();
		// disruptor.shutdown();
	}

}
